package com.jch.lx;


import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class SplitStream {

    //分流
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = env.readTextFile("");

        SingleOutputStreamOperator<Student> map = source.map(new MapFunction<String, Student>() {
            @Override
            public Student map(String s) throws Exception {

                String[] split = s.split(",");

                return new Student(split[0],Integer.parseInt(split[1]));
            }
        });


        //定义两个标签
        OutputTag<Student> tag_even = new OutputTag<Student>("high") {};
        OutputTag<Student> tag_odd = new OutputTag<Student>("low") {};

        //对ds的数据进行处理
        SingleOutputStreamOperator<Student> tagResult = map.process(new ProcessFunction<Student, Student>() {
            @Override
            public void processElement(Student student, Context context, Collector<Student> collector) throws Exception {
                if (student.getAge() > 30) {
                    context.output(tag_even, student);
                }
            }
        });

        //取出标记好的数据
        DataStream<Student> eventResult = tagResult.getSideOutput(tag_even);
        DataStream<Student> oddResult = tagResult.getSideOutput(tag_odd);

        //合流
        ConnectedStreams<Student, Student> connect = eventResult.connect(oddResult);
        SingleOutputStreamOperator<Object> streamOperator = connect.map(new CoMapFunction<Student, Student, Object>() {

            @Override
            public Object map1(Student student) throws Exception {
                return null;
            }

            @Override
            public Object map2(Student student) throws Exception {
                return null;
            }
        });


    }


}
